package io.github.wanggit.antrpc.client.zk.listener;

import io.github.wanggit.antrpc.client.zk.IZkClient;
import io.github.wanggit.antrpc.client.zk.lb.ILoadBalancerContainer;
import io.github.wanggit.antrpc.client.zk.zknode.IReportSubscriber;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.lb.LoadBalancerDTO;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.zookeeper.ZkNodeType;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;

@Slf4j
public class LbNodeChangeListener implements IListener {

    private final IZkClient zkClient;
    private final ILoadBalancerContainer loadBalancerContainer;
    private final IReportSubscriber reportSubscriber;

    public LbNodeChangeListener(
            IZkClient zkClient,
            ILoadBalancerContainer loadBalancerContainer,
            IReportSubscriber reportSubscriber) {
        this.zkClient = zkClient;
        this.loadBalancerContainer = loadBalancerContainer;
        this.reportSubscriber = reportSubscriber;
    }

    @Override
    public void listen() {
        CuratorFramework curator = zkClient.getCurator();
        internalListen(curator);
    }

    /**
     *
     *
     * <pre>
     * <[CDATA[
     * /__rpc_lbs__/io.github.wanggit.antrpc.demo.api.HelloService
     * {
     * 	"lb": "io.github.wanggit.antrpc.client.zk.lb.RandomLoadBalancer",
     * 	"clazz": "io.github.wanggit.antrpc.demo.api.HelloService"
     * 	"weights": {
     * 		"127.0.0.1:8089" : 5
     * 		"127.0.0.1:8088" : 1
     * 	}
     * }
     * ]]>
     * </pre>
     *
     * @param curator zkclient
     */
    private void internalListen(CuratorFramework curator) {
        TreeCache lbNodeTreeCache =
                new TreeCache(curator, "/" + ConstantValues.ZK_ROOT_LB_NODE_NAME);
        lbNodeTreeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            ChildData childData = event.getData();
                            if (null == childData) {
                                return;
                            }
                            String path = childData.getPath();
                            ZkNodeType.LbNodeType lbNodeType = ZkNodeType.getLbNodeType(path);
                            if (ZkNodeType.LbNodeType.INTERFACE.equals(lbNodeType)) {
                                LoadBalancerDTO loadBalancerDTO =
                                        JsonUtils.fromJson(
                                                childData.getData(), LoadBalancerDTO.class);
                                // 只有本服务订阅的接口才处理本地信息
                                if (reportSubscriber.wasSubscribed(loadBalancerDTO.getClazz())) {
                                    if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType())) {
                                        loadBalancerContainer.addLoadBalancer(loadBalancerDTO);
                                    } else if (TreeCacheEvent.Type.NODE_UPDATED.equals(
                                            event.getType())) {
                                        loadBalancerContainer.updateLoadBalancer(loadBalancerDTO);
                                    } else if (TreeCacheEvent.Type.NODE_REMOVED.equals(
                                            event.getType())) {
                                        loadBalancerContainer.deleteLoadBalancer(
                                                loadBalancerDTO.getClazz());
                                    }
                                }
                            }
                        });

        try {
            lbNodeTreeCache.start();
        } catch (Exception e) {
            String errorMessage = "TreeCache failed to start.";
            if (log.isErrorEnabled()) {
                log.error(errorMessage, e);
            }
            throw new ZkListenerException(errorMessage, e);
        }
    }
}
